## Casting Salaries
df = df.withColumn("SALARY_FROM", col("SALARY_FROM").cast("float")) \
.withColumn("SALARY_TO", col("SALARY_TO").cast("float")) \
.withColumn("SALARY", col("SALARY").cast("float")) \
.withColumn("MIN_YEARS_EXPERIENCE", col("MIN_YEARS_EXPERIENCE").cast("float")) \
.withColumn("MAX_YEARS_EXPERIENCE", col("MAX_YEARS_EXPERIENCE").cast("float")) \
.withColumn("EDUCATION_LEVELS_NAME",regexp_replace(col("EDUCATION_LEVELS_NAME"), r"[\n\r]", "")) \
## Computing Medians
def compute_median(sdf, col_name):
q = sdf.approxQuantile(col_name, [0.5], 0.01)
return q[0] if q else None
median_from = compute_median(df, "SALARY_FROM")
median_to = compute_median(df, "SALARY_TO")
median_salary = compute_median(df, "SALARY")
print("Medians:", median_from, median_to, median_salary)
## Imput missing using the medians
df = df.fillna({
"SALARY_FROM": median_from,
"SALARY_TO": median_to,
"SALARY": median_salary
})
## compute average salary
df = df.withColumn("Average_Salary", (col("SALARY_FROM") + col("SALARY_TO"))/2)
## removing null values in Employmet type column
df = df.na.drop(subset=["EMPLOYMENT_TYPE_NAME"])
# df.select("Average_Salary", "SALARY", "EDUCATION_LEVELS_NAME", "REMOTE_TYPE_NAME", "MAX_YEARS_EXPERIENCE", "LOT_V6_SPECIALIZED_OCCUPATION_NAME").show(5, truncate=False)
## selecting required columns & exporting data/ saving to csv
export_cols = [
"EDUCATION_LEVELS_NAME",
"REMOTE_TYPE_NAME",
"MAX_YEARS_EXPERIENCE",
"Average_Salary",
"SALARY_TO",
"SALARY_FROM",
"SALARY",
"LOT_V6_SPECIALIZED_OCCUPATION_NAME",
"LOT_OCCUPATION_NAME",
"NAICS2_NAME",
"EMPLOYMENT_TYPE_NAME",
"MIN_YEARS_EXPERIENCE"
]
df_selected = df.select(*export_cols)
## export
pdf = df_selected.toPandas()
pdf.to_csv("lightcast_cleaned.csv", index=False)
#removing random characters from these columns
pdf["EMPLOYMENT_TYPE_NAME"] = pdf["EMPLOYMENT_TYPE_NAME"].astype(str).apply(
lambda x: re.sub(r"[^\x00-\x7F]+", "", x)
)
pdf["EDUCATION_LEVELS_NAME"] = pdf["EDUCATION_LEVELS_NAME"].astype(str).str.replace(r"[\n\r\\\"\[\]]", "", regex=True)
print(pdf.columns.tolist())
print("Data cleaning complete. Row retained:", len(pdf))